package edu.utexas.mapreduce;

import java.io.IOException;
import java.io.InputStream;
import java.io.PrintWriter;
import java.io.StringReader;
import java.io.StringWriter;
import java.util.Map;

import org.xml.sax.InputSource;
import org.xml.sax.SAXException;
import org.xml.sax.XMLReader;

import com.sun.org.apache.xerces.internal.parsers.SAXParser;

public class Job {

	//configuration members and respective keys	
	public String input=null;
	public static final String inputKey = "_job_input";
	public String output=null;
	public static final String outputKey = "_job_output";	
	public String name=null;
	public static final String nameKey = "_job_name";
	public String jar=null;
	public static final String jarKey = "_job_jar";
	public String map=null;
	public static final String mapKey = "_job_map";
	public String reduce=null;
	public static final String reduceKey = "_job_reduce";
	public String requesterName=null;
	public static final String requesterNameKey = "_job_requester_name";
	public String requesterEmail=null;
	public static final String requesterEmailKey = "_job_requester_email";
	public String jobURL=null;
	public static final String jobURLKey = "_job_url";
	public static final String jobKey = "_job";
	private Map<String, String> keyMap;
	//internal XML fields for job relay operations
	public String jobIsRelay=null;
	public static final String jobIsRelayKey = "_job_isrelay";
	public String jobId=null;
	public static final String jobIdKey = "_job_id";
	public String jobSplits=null;
	public static final String jobSplitsKey = "_job_splits";
	
	//Constructors
	//When creating from scratch
	public Job(){
		keyMap=null;
	}
	
	//When creating from a map with key-value pairs
	public Job(Map<String, String> jobMap){
		keyMap=jobMap;
		jobMapToMembers ();
	}
	
	//When creating from an XML string
	public  Job (String jobXMLString) throws IOException{
		XMLReader myXMLReader = new SAXParser();
		try {
			//parse XML to map
			JobContentHandler contentHandler = new JobContentHandler();	
			myXMLReader.setContentHandler(contentHandler);
			myXMLReader.parse(new InputSource(new StringReader(jobXMLString)));
			keyMap = contentHandler.getRequest().elementAt(0);
			jobMapToMembers ();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (SAXException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	//When creating from an XML stream
	public  Job (InputStream jobXMLStream) throws IOException{
		XMLReader myXMLReader = new SAXParser();
		try {
			//parse XML to map
			JobContentHandler contentHandler = new JobContentHandler();	
			myXMLReader.setContentHandler(contentHandler);
			myXMLReader.parse(new InputSource(jobXMLStream));
			keyMap = contentHandler.getRequest().elementAt(0);
			jobMapToMembers ();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (SAXException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	private void jobMapToMembers (){
		//assign values to members
		if (keyMapValue(inputKey).equals(""))
			input = null;
		else
			input = keyMapValue(inputKey);
		if (keyMapValue(outputKey).equals(""))
			output = null;
		else
			output = keyMapValue(outputKey);
		if (keyMapValue(jarKey).equals(""))		
			jar = null;
		else	
			jar = keyMapValue(jarKey);
		if (keyMapValue(nameKey).equals(""))
			name = null;
		else	
			name = keyMapValue(nameKey);
		if (keyMapValue(mapKey).equals(""))
			map = null;
		else
			map = keyMapValue(mapKey);
		if (keyMapValue(reduceKey).equals(""))
			reduce = null;
		else	
			reduce = keyMapValue(reduceKey);
		if (keyMapValue(requesterNameKey).equals(""))
			requesterName = null;
		else
			requesterName = keyMapValue(requesterNameKey);
		if (keyMapValue(requesterEmailKey).equals(""))
			requesterEmail = null;
		else
			requesterEmail = keyMapValue(requesterEmailKey);
		if (keyMapValue(jobURLKey).equals(""))
			jobURL = null;
		else
			jobURL = keyMapValue(jobURLKey);
		if (keyMapValue(jobIdKey).equals(""))
			jobId = null;
		else
			jobId= keyMapValue(jobIdKey);
		if (keyMapValue(jobIsRelayKey).equals(""))
			jobIsRelay = null;
		else
			jobIsRelay = keyMapValue(jobIsRelayKey);
		if (keyMapValue(jobSplitsKey).equals(""))
			jobSplits = null;
		else	
			jobSplits = keyMapValue(jobSplitsKey);
	}
			
	private String keyMapValue (String key){
		if (keyMap.containsKey(key) && (keyMap.get(key) != null))
			return keyMap.get(key).trim();
		else
			return "";
	}

	
	public boolean validate(){
		//validate that the required fields exist
		if (input==null)
			return false;
		if (output==null)
			return false;
		if (map==null)
			return false;
		if (reduce==null)
			return false;
		if (jobURL==null)
			return false;
		return true;
	}
	
	@Override
	public String toString(){
		StringWriter sw = new StringWriter();
		PrintWriter out = new PrintWriter(sw);
		printMemberAsLine(out, "jar", jar);
		printMemberAsLine(out, "map", map);
		printMemberAsLine(out, "reduce", reduce);
		printMemberAsLine(out, "input", input);
		printMemberAsLine(out, "output", output);
		printMemberAsLine(out, "url", jobURL);
		printMemberAsLine(out, "requester", requesterName);
		printMemberAsLine(out, "email", requesterEmail);
		printMemberAsLine(out, "is relay", jobIsRelay);
		printMemberAsLine(out, "splits", jobSplits);
		printMemberAsLine(out, "job id", jobId);		
		
		out.flush();
		sw.flush();
		return sw.toString();
	}
	
	private void printMemberAsLine (PrintWriter pw, String name, String value){
		if (value!=null){
			pw.printf("  %.12s : %s\n", name, value); 
			//pw.println(nameString + value);
		}
	}
	
	
	public String toXMLString(boolean addHeader){
		
		// Write XML header and DTD		
		int indentation = 0;
		StringWriter sw = new StringWriter();
		PrintWriter out = new PrintWriter(sw);
		if (addHeader)
			out.println("<?xml version=\"1.0\" encoding=\"" + XML.ENCODING + "\"?>");
		out.println("<job>");
		indentation +=4;
		out.println(printXMLElement(nameKey, name, indentation));
		out.println(printXMLElement(inputKey, input, indentation));
		out.println(printXMLElement(outputKey, output, indentation));
		out.println(printXMLElement(mapKey, map, indentation));
		out.println(printXMLElement(reduceKey, reduce, indentation));
		out.println(printXMLElement(jarKey, jar, indentation));
		out.println(printXMLElement(jobURLKey, jobURL, indentation));
		out.println(printXMLElement(jobIsRelayKey, jobIsRelay, indentation));
		out.println(printXMLElement(jobIdKey, jobId, indentation));
		out.println(printXMLElement(jobSplitsKey, jobSplits, indentation));
		out.println("<requester>");
		indentation +=4;
		out.println(printXMLElement(requesterNameKey, requesterName, indentation));
		out.println(printXMLElement(requesterEmailKey, requesterEmail, indentation));
		indentation -=4;
		out.println("</requester>");
		indentation -=4;
		out.println("</job>");
		out.flush();
		
		return sw.toString();
	}
	
	private String printXMLElement(String name, String value, int indentation){
		String keys[] = name.split("_");
		String key = keys[keys.length-1];
		if (value == null)
			return "<" + key + "></" + key + ">"; 
		else
			return "<" + key + ">" + value + "</" + key + ">"; 
	}

	@Override
	public boolean equals(Object obj) {
		if (obj.getClass()!=this.getClass())
			return false;
		if (!fieldEquality(((Job) obj).input, this.input))
			return false;
		if (!fieldEquality(((Job) obj).output, this.output))
			return false;
		if (!fieldEquality(((Job) obj).map, this.map))
			return false;
		if (!fieldEquality(((Job) obj).reduce, this.reduce))
			return false;
		if (!fieldEquality(((Job) obj).jar, this.jar))
			return false;
		if (!fieldEquality(((Job) obj).requesterName, this.requesterName))
			return false;
		if (!fieldEquality(((Job) obj).requesterEmail, this.requesterEmail))
			return false;
		if (!fieldEquality(((Job) obj).jobURL, this.jobURL))
			return false;
		if (!fieldEquality(((Job) obj).jobId, this.jobId))
			return false;
		return true;
	}
	
	private boolean fieldEquality (Object obj1, Object obj2){
		if (obj1 ==null && obj2 == null) //if both null, they are considered equals
			return true;
		if (obj1 ==null ^ obj2 == null) //if only one of them is null, they are different
			return false;
		if (((String) obj1).equals((String) obj2)) //if their values match
			return true;
		return false;
	}
	 
}
